

namespace Stee.CAP8.MessageBus.Kafka.Utility
{
    using System;
    using System.Collections.Generic;
    using System.Linq;
    using Microsoft.Extensions.Configuration;
    using NLog;
    using Stee.CAP8.MessageBus.Kafka.Constants;

    /// <summary>
    /// Helper for reading configurations for broker & topics
    /// </summary>
    public static class ConfigurationHelper
    {
        private readonly static Logger Logger = LogManager.GetCurrentClassLogger();


        /// <summary>
        /// Default configuration for dead-letter topics
        /// </summary>
        private readonly static IDictionary<string, string> DLQTopicConfiguration = new Dictionary<string, string>();


        /// <summary>
        /// Initializes the static ConfigurationHelper class
        /// </summary>
        static ConfigurationHelper()
        {
            //// Read in dead-letter queue configuration
            var dlqConfigs = ReadTopicConfiguration(
                ConfigurationConstants.DefaultTopicConfigurationJSON, ConfigurationConstants.TopicDLQConfigSection);

            if (dlqConfigs == null)
            {
                Logger.Warn($"Default topic configuration file is invalid: {ConfigurationConstants.DefaultTopicConfigurationJSON}. Default topic configuration will be used.");
            }
            else
            {
                foreach (var item in dlqConfigs)
                {
                    //// Use Add(), allow duplicate keys to throw exception
                    DLQTopicConfiguration.Add(item.Key, item.Value);
                }
            }
        }


        #region Broker Configuration

        /// <summary>
        /// Get default configuration with specified sector.
        /// </summary>
        /// <param name="configSection">Configuration section</param>
        /// <returns>Return collection mapping of default configurations</returns>
        public static IDictionary<string, string> GetDefaultBrokerConfiguration(string configSection) =>
            GetBrokerConfiguration(ConfigurationConstants.DefaultBrokerConfigurationJSON, configSection);

        /// <summary>
        /// Read configuration from JSON file with specified sector.
        /// If keys are duplicated in JSON file, the last key will over-write previous entry. 
        /// </summary>
        /// <param name="jsonConfig">JSON configuration file path</param>
        /// <param name="configSection">Configuration section</param>
        /// <returns>Return collection mapping of configurations if file is valid else NULL</returns>
        public static IDictionary<string, string> GetBrokerConfiguration(string jsonConfig, string configSection)
        {
            try
            {
                if (IsJsonConfigSectionExist(jsonConfig, configSection))
                {
                    var configurationRoot = new ConfigurationBuilder().AddJsonFile(jsonConfig, false).Build();
                    var sectionMessageBus = configurationRoot.GetSection(configSection);
                    var section = sectionMessageBus.GetChildren().ToList();
                    var configs = new Dictionary<string, string>();

                    foreach (var item in section)
                    {
                        var key = item.Key.Trim();
                        var value = item.Value.Trim();

                        configs[key] = value;
                    }

                    return configs;
                }
                else
                {
                    return new Dictionary<string, string>();
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, $"Error on reading JSON application sector, {configSection}. Exception: {ex.Message}");
                throw;
            }
        }

        #endregion Broker Configuration


        #region Topic Configuration

        /// <summary>
        /// Read multiple topic configurations from JSON file with specified sector.
        /// If keys are duplicated in JSON file, the last key will over-write previous entry. 
        /// </summary>
        /// <param name="jsonConfig">JSON topic configuration file path</param>
        /// <param name="configSection">Configuration section</param>
        /// <returns>Return collection mapping of configurations if configuration is found else NULL</returns>
        public static IList<IDictionary<string, string>> ReadMultiTopicConfiguration(string jsonConfig,
            string configSection = ConfigurationConstants.TopicConfigSection)
        {
            try
            {
                if (IsJsonConfigSectionExist(jsonConfig, configSection))
                {
                    var configurationRoot = new ConfigurationBuilder().AddJsonFile(jsonConfig, false).Build();
                    var sectionMessageBus = configurationRoot.GetSection(configSection);
                    var section = sectionMessageBus.GetChildren().ToList();
                    var configs = new List<IDictionary<string, string>>();

                    //// Get topic
                    foreach (var item in section)
                    {
                        var topic = item.GetChildren().ToList();
                        var topicCfg = new Dictionary<string, string>();

                        //// Get topic configs
                        foreach (var cfg in topic)
                        {
                            topicCfg[cfg.Key] = cfg.Value;
                        }

                        configs.Add(topicCfg);
                    }

                    return configs;
                }
                else
                {
                    return new List<IDictionary<string, string>>();
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, $"Error on reading JSON application sector, {configSection}. Exception: {ex.Message}");
                throw;
            }
        }

        /// <summary>
        /// Read topic configurations from JSON file with specified sector.
        /// </summary>
        /// <param name="jsonConfig">JSON topic configuration file path</param>
        /// <param name="configSection">Configuration section</param>
        /// <returns>Return dead-letter queue configuration if found else NULL</returns>
        public static IDictionary<string, string> ReadTopicConfiguration(string jsonConfig, string configSection)
        {
            try
            {
                if (IsJsonConfigSectionExist(jsonConfig, configSection))
                {
                    var configurationRoot = new ConfigurationBuilder().AddJsonFile(jsonConfig, false).Build();
                    var sectionMessageBus = configurationRoot.GetSection(configSection);
                    var section = sectionMessageBus.GetChildren().ToList();
                    var topicCfg = new Dictionary<string, string>();

                    foreach (var item in section)
                    {
                        topicCfg[item.Key] = item.Value;
                    }

                    return topicCfg;
                }
                else
                {
                    return new Dictionary<string, string>();
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, $"Error on reading JSON application sector, {configSection}. Exception: {ex.Message}");
                throw;
            }
        }

        /// <summary>
        /// Gets a new instance of the dead-letter queue configuration
        /// </summary>
        /// <returns>New dead-letter queue configuration instance</returns>
        public static IDictionary<string, string> GetDLQTopicConfiguration()
        {
            var config = new Dictionary<string, string>();

            foreach (var item in DLQTopicConfiguration)
            {
                config.Add(item.Key, item.Value);
            }

            return config;
        }

        /// <summary>
        /// Generate new dead letter queue topic name from specified topic name
        /// </summary>
        /// <param name="topic">Topic name</param>
        /// <returns>Return new dead letter queue topic name</returns>
        public static string GenerateDLQTopicName(string topic)
        {
            var topicStr = topic?.Trim();

            if (string.IsNullOrEmpty(topicStr))
            {
                throw new ArgumentNullException($"Null topic encountered: {nameof(topic)}");
            }

            //// Check if topic already has DLQ extension to prevent "<topic>_dlq_dlq"
            return CheckDLQTopicName(topicStr) ?
                topicStr : $"{topicStr}{ConfigurationConstants.TopicNamePostFixDLQ}";
        }

        /// <summary>
        /// Check if topic name ends with dead-letter queue extension.
        /// If name matches the extension exactly, result is FALSE.
        /// </summary>
        /// <param name="topicName">Topic name</param>
        /// <returns>Return TRUE if topic name ends with dead-letter queue extension else FALSE</returns>
        public static bool CheckDLQTopicName(string topicName)
        {
            var topicStr = topicName?.Trim();

            if (string.IsNullOrEmpty(topicStr))
            {
                throw new ArgumentNullException($"Null topic encountered: {nameof(topicName)}");
            }

            //// Return TRUE if topic is minimally "<topic>_dlq"
            return (topicStr.Length > ConfigurationConstants.TopicNamePostFixDLQ.Length) &&
                topicStr.EndsWith(ConfigurationConstants.TopicNamePostFixDLQ, StringComparison.InvariantCultureIgnoreCase);
        }

        #endregion Topic Configuration


        #region Validate Configuration File

        /// <summary>
        /// Check for section existance in JSON configuration file
        /// </summary>
        /// <param name="jsonFilePath">JSON configuration file path</param>
        /// <param name="configSection">Configuration section</param>
        /// <returns>Return TRUE if the section is found else FALSE</returns>
        public static bool IsJsonConfigSectionExist(string jsonFilePath, string configSection)
        {
            if (string.IsNullOrEmpty(jsonFilePath))
            {
                throw new ArgumentNullException($"Configuration file path cannot be empty: {nameof(jsonFilePath)}");
            }
            else if (string.IsNullOrEmpty(configSection))
            {
                throw new ArgumentNullException($"Configuration sector name cannot be empty: {nameof(configSection)}");
            }
            else
            {
                var configurationRoot = new ConfigurationBuilder().AddJsonFile(jsonFilePath, false).Build();
                var section = configurationRoot.GetSection(configSection);

                return section?.Exists() ?? false;
            }
        }

        #endregion Validate Configuration File


        /* Sample allowable configurations w/ default values
         * https://docs.confluent.io/current/installation/configuration/broker-configs.html
         * ===============================
         * log.cleaner.min.compaction.lag.ms            					0
         * offsets.topic.num.partitions            							50
         * log.flush.interval.messages            							9223372036854775807
         * controller.socket.timeout.ms            							30000
         * principal.builder.class            								""
         * log.flush.interval.ms            								""
         * min.insync.replicas            									1
         * num.recovery.threads.per.data.dir            					1
         * ssl.keystore.type            									JKS
         * zookeeper.ssl.protocol            								TLSv1.2
         * sasl.mechanism.inter.broker.protocol            					GSSAPI
         * fetch.purgatory.purge.interval.requests         					1000
         * ssl.endpoint.identification.algorithm           					https
         * zookeeper.ssl.keystore.location            	                    ""
         * replica.socket.timeout.ms            							30000
         * message.max.bytes            									1048588
         * connections.max.reauth.ms            							0
         * log.flush.offset.checkpoint.interval.ms            				60000
         * zookeeper.clientCnxnSocket            				            ""
         * zookeeper.ssl.client.enable            							false
         * quota.window.num            										11
         * zookeeper.connect            									localhost:2181
         * authorizer.class.name            				                ""
         * password.encoder.secret            				                ""
         * num.replica.fetchers            									1
         * alter.log.dirs.replication.quota.window.size.seconds            	1
         * log.roll.jitter.hours            								0
         * password.encoder.old.secret            	                        ""
         * log.cleaner.delete.retention.ms            						86400000
         * queued.max.requests            									500
         * log.cleaner.threads            									1
         * sasl.kerberos.service.name            	                        ""
         * socket.request.max.bytes            								104857600
         * log.message.timestamp.type            							CreateTime
         * zookeeper.ssl.keystore.type            	                        ""
         * connections.max.idle.ms            								600000
         * zookeeper.set.acl            									false
         * delegation.token.expiry.time.ms            						86400000
         * max.connections            										2147483647
         * transaction.state.log.num.partitions            					50
         * listener.security.protocol.map            						PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
         * log.retention.hours            									168
         * client.quota.callback.class            	                        ""
         * ssl.provider            	                                        ""
         * delete.records.purgatory.purge.interval.requests            		1
         * log.roll.ms            	                                        ""
         * ssl.cipher.suites            	                                ""
         * zookeeper.ssl.keystore.password            	                    ""
         * security.inter.broker.protocol            						PLAINTEXT
         * replica.high.watermark.checkpoint.interval.ms            		5000
         * replication.quota.window.size.seconds            				1
         * sasl.kerberos.ticket.renew.window.factor            				0.8
         * zookeeper.connection.timeout.ms            						18000
         * metrics.recording.level            								INFO
         * password.encoder.cipher.algorithm            					AES/CBC/PKCS5Padding
         * ssl.principal.mapping.rules            							DEFAULT
         * replica.selector.class            				                ""
         * max.connections.per.ip            								2147483647
         * background.threads            									10
         * quota.consumer.default            								9223372036854775807
         * request.timeout.ms            									30000
         * log.message.format.version            							2.5-IV0
         * sasl.login.class            				                        ""
         * log.dir            												/tmp/kafka-logs
         * log.segment.bytes            									1073741824
         * replica.fetch.response.max.bytes            						10485760
         * group.max.session.timeout.ms            							1800000
         * port            													9092
         * log.segment.delete.delay.ms            							60000
         * log.retention.minutes            				                ""
         * log.dirs            												""                          *Must be recommended to set
         * controlled.shutdown.enable            							true
         * log.message.timestamp.difference.max.ms            				9223372036854775807
         * password.encoder.key.length            							128
         * sasl.login.refresh.min.period.seconds            				60
         * transaction.abort.timed.out.transaction.cleanup.interval.ms      10000
         * sasl.kerberos.kinit.cmd            								/usr/bin/kinit
         * log.cleaner.io.max.bytes.per.second            					1.7976931348623157E308
         * auto.leader.rebalance.enable            							true
         * leader.imbalance.check.interval.seconds            				300
         * log.cleaner.min.cleanable.ratio            						0.5
         * replica.lag.time.max.ms            								30000
         * num.network.threads            									3
         * sasl.client.callback.handler.class            				    ""
         * metrics.num.samples            									2
         * socket.send.buffer.bytes            								102400
         * password.encoder.keyfactory.algorithm            				""
         * socket.receive.buffer.bytes            							102400
         * replica.fetch.min.bytes            								1
         * broker.rack            				                            ""
         * zookeeper.ssl.truststore.password            				    ""
         * unclean.leader.election.enable            						false
         * offsets.retention.check.interval.ms            					600000
         * producer.purgatory.purge.interval.requests          				1000
         * metrics.sample.window.ms            								30000
         * log.retention.check.interval.ms            						300000
         * sasl.login.refresh.window.jitter            						0.05
         * leader.imbalance.per.broker.percentage            				10
         * advertised.host.name            				                    ""
         * metric.reporters            				                        ""
         * quota.producer.default            								9223372036854775807
         * auto.create.topics.enable            							true
         * replica.socket.receive.buffer.bytes            					65536
         * zookeeper.ssl.truststore.location            				    ""
         * replica.fetch.wait.max.ms            							500
         * password.encoder.iterations            							4096
         * default.replication.factor            							1
         * ssl.truststore.password            				                ""
         * sasl.kerberos.principal.to.local.rules            				DEFAULT
         * log.preallocate            										false
         * transactional.id.expiration.ms            						604800000
         * control.plane.listener.name            				            ""
         * transaction.state.log.replication.factor            				1
         * num.io.threads                                                   8
         * sasl.login.refresh.buffer.seconds            					300
         * offsets.commit.required.acks            							-1
         * connection.failed.authentication.delay.ms           				100
         * delete.topic.enable            									true
         * quota.window.size.seconds            							1
         * ssl.truststore.type            									JKS
         * offsets.commit.timeout.ms            							5000
         * zookeeper.ssl.ocsp.enable            							false
         * log.cleaner.max.compaction.lag.ms            					9223372036854775807
         * zookeeper.ssl.enabled.protocols                                  ""
         * log.retention.ms                                                 ""
         * alter.log.dirs.replication.quota.window.num            			11
         * log.cleaner.enable            									true
         * offsets.load.buffer.size            								5242880
         * ssl.client.auth            										none
         * controlled.shutdown.max.retries            						3
         * offsets.topic.replication.factor            						1
         * zookeeper.ssl.truststore.type                                    ""
         * transaction.state.log.min.isr            						1
         * ssl.secure.random.implementation                                 ""
         * sasl.kerberos.ticket.renew.jitter            					0.05
         * ssl.trustmanager.algorithm            							PKIX
         * zookeeper.session.timeout.ms            							18000
         * log.retention.bytes            									-1
         * sasl.jaas.config                                                 ""
         * sasl.kerberos.min.time.before.relogin            				60000
         * offsets.retention.minutes            							10080
         * replica.fetch.backoff.ms            								1000
         * inter.broker.protocol.version            						2.5-IV0                     *Not recommended to set
         * kafka.metrics.reporters                                          ""
         * num.partitions            										1
         * broker.id.generation.enable            							true
         * listeners                                                        ""
         * ssl.enabled.protocols            								TLSv1.2
         * inter.broker.listener.name                                       ""
         * alter.config.policy.class.name                                   ""
         * delegation.token.expiry.check.interval.ms            			3600000
         * log.flush.scheduler.interval.ms            						9223372036854775807
         * zookeeper.max.in.flight.requests            						10
         * log.index.size.max.bytes            								10485760
         * ssl.keymanager.algorithm            								SunX509
         * sasl.login.callback.handler.class   								""   
         * replica.fetch.max.bytes            								1048576
         * zookeeper.ssl.crl.enable            								false
         * sasl.server.callback.handler.class            		            ""
         * log.cleaner.dedupe.buffer.size            						134217728
         * advertised.port            		                                ""
         * log.cleaner.io.buffer.size            							524288
         * create.topic.policy.class.name            		                ""
         * controlled.shutdown.retry.backoff.ms            					5000
         * security.providers            		                            ""
         * log.roll.hours            										168
         * log.cleanup.policy            									delete
         * log.flush.start.offset.checkpoint.interval.ms            		60000
         * host.name            		                                    ""
         * log.roll.jitter.ms            		                            ""
         * transaction.state.log.segment.bytes            					104857600
         * offsets.topic.segment.bytes            							104857600
         * group.initial.rebalance.delay.ms            						0
         * log.index.interval.bytes            								4096
         * log.cleaner.backoff.ms            								15000
         * ssl.truststore.location            						        ""
         * offset.metadata.max.bytes            							4096
         * ssl.keystore.password            						        ""
         * zookeeper.sync.time.ms            								2000
         * fetch.max.bytes            										57671680
         * compression.type            										producer
         * max.connections.per.ip.overrides            		                ""
         * sasl.login.refresh.window.factor            						0.8
         * kafka.metrics.polling.interval.secs            					10
         * max.incremental.fetch.session.cache.slots            			1000
         * delegation.token.master.key            		                    ""
         * ssl.key.password            		                                ""
         * reserved.broker.max.id            								1000
         * transaction.remove.expired.transaction.cleanup.interval.ms       3600000
         * log.message.downconversion.enable            					true
         * ssl.protocol            											TLSv1.2
         * transaction.state.log.load.buffer.size            				5242880
         * ssl.keystore.location            			                    ""
         * sasl.enabled.mechanisms            								GSSAPI
         * num.replica.alter.log.dirs.threads            			        ""
         * zookeeper.ssl.cipher.suites            			                ""
         * group.min.session.timeout.ms            							6000
         * log.cleaner.io.buffer.load.factor            					0.9
         * transaction.max.timeout.ms            							900000
         * group.max.size            										2147483647
         * delegation.token.max.lifetime.ms            						604800000
         * broker.id            											0
         * offsets.topic.compression.codec            						0
         * zookeeper.ssl.endpoint.identification.algorithm            		HTTPS
         * replication.quota.window.num            							11
         * advertised.listeners                                             ""
         * queued.max.request.bytes            								-1
         */


        /* Sample allowable configurations w/ default values
         * https://docs.confluent.io/current/installation/configuration/topic-configs.html
         * ===============================
         * compression.type    						    producer
         * leader.replication.throttled.replicas   	    ""
         * message.downconversion.enable    			true
         * min.insync.replicas    						1
         * segment.jitter.ms    						0
         * cleanup.policy    							delete
         * flush.ms    								    9223372036854775807
         * follower.replication.throttled.replicas      ""
         * segment.bytes    							1073741824
         * retention.ms    							    604800000
         * flush.messages    							9223372036854775807
         * message.format.version    					2.5-IV0                     *Not recommended to set
         * max.compaction.lag.ms    					9223372036854775807
         * file.delete.delay.ms    					    60000
         * max.message.bytes    						1048588
         * min.compaction.lag.ms    					0
         * message.timestamp.type    					CreateTime
         * preallocate    								false
         * min.cleanable.dirty.ratio   				    0.5
         * index.interval.bytes    					    4096
         * unclean.leader.election.enable    			false
         * retention.bytes    							-1
         * delete.retention.ms    						86400000
         * segment.ms    								604800000
         * message.timestamp.difference.max.ms    		9223372036854775807
         * segment.index.bytes    						10485760
         */
    }
}
